package com.shengzai.flink.core

import org.apache.flink.streaming.api.scala._

object Demo1WorldCountStream {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置并行度
    env.setParallelism(2)

    //设置发送时间
    env.setBufferTimeout(200)

    val DS: DataStream[String] = env.socketTextStream("master", 8888)

    val flatDS: DataStream[String] = DS.flatMap(
      line => {
        line.split(",")
      }
    )

    val mapDS: DataStream[(String, Int)] = flatDS.map(
      word => {
        (word, 1)
      }
    )

    val keyByDS: KeyedStream[(String, Int), String] = mapDS.keyBy(
      word => {
        word._1
      }
    )

    val resDS: DataStream[(String, Int)] = keyByDS.sum(1)

    resDS.print()

    env.execute()



  }

}
